package test.rocketmq;

import java.util.concurrent.TimeUnit;

import org.apache.jmeter.config.Arguments;
import org.apache.jmeter.protocol.java.sampler.AbstractJavaSamplerClient;
import org.apache.jmeter.protocol.java.sampler.JavaSamplerContext;
import org.apache.jmeter.samplers.SampleResult;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;

public class RocketMQConsumer extends AbstractJavaSamplerClient {
	
	

	
	private String nameServer ;
	private int receiveCount;
	
	
	
	
	@Override
	public Arguments getDefaultParameters() {
		Arguments params = new Arguments();
		params.addArgument("receiveCount","");
		params.addArgument("server:port","");
		
		return params;
	}
	
	@Override
	public void setupTest(JavaSamplerContext context) {
		if (context.containsParameter("server:port")) {
			this.nameServer = context.getParameter("server:port");
		}
		if (context.containsParameter("receiveCount")) {
			this.receiveCount = Integer.valueOf(context.getParameter("receiveCount"));
		}
	}
	

	@Override
	public SampleResult runTest(JavaSamplerContext context) {
		SampleResult sr = new SampleResult();
		sr.setSampleLabel("RocketMQ消费者测试");
		try {
			sr.sampleStart();
			DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MixUtil.createGroupName("ConsumerGroupname_"));
			consumer.setNamesrvAddr(nameServer);
			consumer.setInstanceName("Consumer");
			consumer.setConsumeMessageBatchMaxSize(receiveCount);  //设置一次接收多少条
			consumer.subscribe("Message1Topic", "*");

			consumer.subscribe("Message2Topic", "*");
			consumer.subscribe("Message3Topic", "*");
			

			consumer.registerMessageListener(new OutMessageListener());

			consumer.start();
			System.out.println("Consumer Started.");

			TimeUnit.MILLISECONDS.sleep(100 * 1000);

			consumer.shutdown();

			sr.setResponseData("success", "utf-8");
			sr.setDataType(SampleResult.TEXT);
			sr.setSuccessful(true);
		} catch (Exception e) {
			sr.setSuccessful(false);
			sr.setResponseData(MixUtil.getStackTrace(e), "utf-8");
			e.printStackTrace();
		} finally {
			sr.sampleEnd();
		}

		return sr;
	}

}
